package com.allenkerr.queue3;

import java.util.List;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.BoundListOperations;
import org.springframework.data.redis.core.RedisConnectionUtils;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.util.CollectionUtils;
import sun.jvm.hotspot.debugger.ThreadAccess;

public class RedisQueue3<T> {

  private String redisKey;

  private RedisTemplate redisTemplate;
  /**
   * 最大阻塞的容量，超过容量将会导致清空旧数据
   */
  private int cap = Short.MAX_VALUE;
  private byte[] rawKey;
  private RedisConnectionFactory connectionFactory;
  /**
   * for blocking
   */
  private RedisConnection redisConnection;
  /**
   * noblocking
   */
  private BoundListOperations<String, T> listOperations;
  /**
   * 基于底层IO阻塞考虑
   */
  private Lock lock = new ReentrantLock();
  /**
   * 异步回调
   */
  private RedisQueueListener listener;
  private Thread listenerThread;

  private boolean isClosed;

  public void setListener(RedisQueueListener listener) {
    this.listener = listener;
  }

  public void setRedisKey(String redisKey) {
    this.redisKey = redisKey;
  }

  public RedisQueue3(RedisTemplate redisTemplate, String redisKey, RedisQueueListener listener) {

    System.out.println("\n---------------------------------");
    System.out.println("RedisQueue3.RedisQueue3");
    System.out.println("---------------------------------\n");

    this.redisTemplate = redisTemplate;
    this.redisKey = redisKey;
    this.listener = listener;
    this.connectionFactory = redisTemplate.getConnectionFactory();
    this.redisConnection = RedisConnectionUtils.getConnection(this.connectionFactory);
    this.rawKey = redisTemplate.getKeySerializer().serialize(redisKey);
    this.listOperations = redisTemplate.boundListOps(redisKey);
    if (listener != null) {
      listenerThread = new Thread() {
        @Override
        public void run() {
          try {
            while (true) {
              //cast exception? you should check.
              T value = takeFromHead();
              //逐个执行
              if (value != null) {
                try {
                  listener.onMessage(value);
                } catch (Exception e) {
                  e.printStackTrace();
                }
              }
            }
          } catch (InterruptedException e) {
            e.printStackTrace();
          }
        }
      };
      listenerThread.setDaemon(true);
      listenerThread.start();
    }
  }

  /**
   * blocking
   * remove and get last item from queue:BRPOP
   *
   * @return
   */
  public T takeFromTail(int timeout) throws InterruptedException {
    lock.lockInterruptibly();
    try {
      List<byte[]> results = redisConnection.bRPop(timeout, rawKey);
      if (CollectionUtils.isEmpty(results)) {
        return null;
      }
      return (T) redisTemplate.getValueSerializer().deserialize(results.get(1));
    } finally {
      lock.unlock();
    }
  }

  public T takeFromTail() throws InterruptedException {
    return takeFromTail(0);
  }

  /**
   * 从队列的头，插入
   */
  public void pushToHead(T value) {
    listOperations.leftPush(value);
  }

  public void pushToTail(T value) {
    listOperations.rightPush(value);
  }

  /**
   * noblocking
   *
   * @return null if no item in queue
   */
  public T removeFromHead() {
    return listOperations.leftPop();
  }

  public T removeFromTail() {
    return listOperations.rightPop();
  }

  /**
   * blocking
   * remove and get first item from queue:BLPOP
   *
   * @return
   */
  public T takeFromHead(int timeout) throws InterruptedException {
    lock.lockInterruptibly();
    try {
      List<byte[]> results = redisConnection.bLPop(timeout, rawKey);
      if (CollectionUtils.isEmpty(results)) {
        return null;
      }
      return (T) redisTemplate.getValueSerializer().deserialize(results.get(1));
    } finally {
      lock.unlock();
    }
  }

  public T takeFromHead() throws InterruptedException {
    return takeFromHead(0);
  }

}